#include <stdarg.h>
#include <unistd.h>

#include "global.h"
#include "net/netglobal.h"
#include "my_memory.h"
#include "net/threadpool.h"
#include "log.h"

// 静态成员初始化
pthread_mutex_t CThreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t CThreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;
bool CThreadPool::m_shutdown = false;// 标记不退出

//构造函数
CThreadPool::CThreadPool()
{
    m_iRunningThreadNum = 0;  //正在运行的线程，开始给个0
    m_iLastEmgTime = 0;       //上次报告线程不够用了的时间；
    m_iRecvMsgQueueCount = 0; //收消息队列
}

//析构函数
CThreadPool::~CThreadPool()
{    
    //资源释放在StopAll()里统一进行

    //接收消息队列中内容释放
    ClearMsgRecvQueue();
}

// 清理接收消息队列函数
void CThreadPool::ClearMsgRecvQueue(){
    char * sTmpMempoint;
    CMemory *p_memory = CMemory::GetInstance();

    // 暂不考虑互斥了，直接全部退出
    while(!m_MsgRecvQueue.empty())
	{
		sTmpMempoint = m_MsgRecvQueue.front();		
		m_MsgRecvQueue.pop_front(); 
		p_memory->FreeMemory(sTmpMempoint);
	}
}

// 创建线程池中所有线程
bool CThreadPool::Create(int threadNum){
    ThreadItem *pNew;
    int err;

    m_iThreadNum = threadNum; //保存要创建的线程数量
    for(int i = 0; i < m_iThreadNum; ++i){
        m_threadVector.push_back(pNew = new ThreadItem(this));
        err = pthread_create(&pNew->_Handle, NULL, ThreadFunc, pNew);
        if(err != 0){
            // 说明创建出错了
            Log_Error(err, "CThreadPool::Create()创建线程%d失败, 返回的错误码为%d!", i, err);
            return false;
        }
    }

    // 保证线程都启动并运行到pthread_cond_wait()，后续才能被唤醒
    std::vector<ThreadItem*>::iterator iter;
lblfor:
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++){
        if((*iter)->ifrunning == false){
            usleep(100 * 1000);// 说明有没有启动起来的线程，休眠100ms
            goto lblfor;
        }
    }
    return true;
}

//线程入口函数，当用pthread_create()创建线程后，这个ThreadFunc()函数都会被立即执行；
void* CThreadPool::ThreadFunc(void* threadData){
    ThreadItem *pThread = static_cast<ThreadItem*>(threadData);

    // 间接获取this指针，因为本函数是静态函数，是不存在this指针的
    CThreadPool *pThreadPoolObj = pThread->_pThis;

    CMemory *p_memory = CMemory::GetInstance();	
    int err;

    pthread_t tid = pthread_self(); //获取线程自身id，以方便调试打印信息等
    while(true){
        err = pthread_mutex_lock(&m_pthreadMutex);// 获取锁
        if(err != 0){
            // 可能会因为一些特殊原因，获取锁异常，报告问题
            Log_Error(err,"CThreadPool::ThreadFunc()中pthread_mutex_lock()失败，返回的错误码为%d!", err);
        }

        while ((pThreadPoolObj->m_MsgRecvQueue.size() == 0) && m_shutdown == false)
        {
            // 说明此时没有任务，线程得进入等待
            if(pThread->ifrunning == false){
                pThread->ifrunning = true;// 标记为true了才允许调用StopAll()
            }
            pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex);
        }

        // 判断是否要关闭线程池了
        if(m_shutdown){
            pthread_mutex_unlock(&m_pthreadMutex); //解锁互斥量
            break;
        }
        
        // 能走到这里，说明有任务处理（此时还在互斥中）
        char *jobbuf = pThreadPoolObj->m_MsgRecvQueue.front();
        pThreadPoolObj->m_MsgRecvQueue.pop_front();
        --pThreadPoolObj->m_iRecvMsgQueueCount;// 收消息队列减一

        // 解锁互斥量，后续操作不需要互斥了
        err = pthread_mutex_unlock(&m_pthreadMutex);
        if(err != 0){
            // 可能会因为一些特殊原因，释放锁异常，报告问题
            Log_Error(err,"CThreadPool::ThreadFunc()中pthread_mutex_unlock()失败，返回的错误码为%d!",err);
        }

        // 正式开始处理
        ++pThreadPoolObj->m_iRunningThreadNum;// 原子自增操作
        g_socket.ThreadRecvProcFunc(jobbuf);// 处理消息（单线程中）
        p_memory->FreeMemory(jobbuf);// 释放消息内存（处理完毕）
        --pThreadPoolObj->m_iRunningThreadNum;// 原子自减操作
    }

    // while结束说明整个程序要结束
    return (void*)0;
}

// 停止所有线程
void CThreadPool::StopAll(){
    if(m_shutdown == true){
        return;// 避免重复调用此函数
    }
    m_shutdown = true;

    // 唤醒所有阻塞在相同信号量中的线程
    int err = pthread_cond_broadcast(&m_pthreadCond);
    if(err != 0)
    {
        Log_Error(err,"CThreadPool::StopAll()中pthread_cond_broadcast()失败，返回的错误码为%d!", err);
        return;
    }

    // 等待每一个线程结束
    std::vector<ThreadItem*>::iterator iter;
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++){
        pthread_join((*iter)->_Handle, NULL);
    }

    // 能走到这里，说明所有线程都结束了，可以开始释放
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++){
        if(*iter){
            delete *iter;
        }
    }
    m_threadVector.clear();// 清空线程
    Log_Info(0, "CThreadPool::StopAll()成功返回，线程池中线程全部正常结束!");
}

// 收到完整消息，放入消息队列
void CThreadPool::InMsgRecvQueueAndSignal(char *buf){
    // 获取锁
    int err = pthread_mutex_lock(&m_pthreadMutex);
    if(err != 0){
        Log_Error(err,"CThreadPool::InMsgRecvQueueAndSignal()pthread_mutex_lock()失败，返回的错误码为%d!", err);
    }

    m_MsgRecvQueue.push_back(buf);// 放入消息队列
    ++m_iRecvMsgQueueCount;// 收消息队列自增

    // 释放锁
    err = pthread_mutex_unlock(&m_pthreadMutex);   
    if(err != 0){
        Log_Error(err,"CThreadPool::InMsgRecvQueueAndSignal()pthread_mutex_unlock()失败，返回的错误码为%d!",err);
    }

    Call();// 唤醒一个线程来干活
}

//来任务了，调一个线程池中的线程下来干活
void CThreadPool::Call(){
    int err = pthread_cond_signal(&m_pthreadCond);// 发条件变量信号
    if(err != 0 ){
        Log_Error(err,"CThreadPool::Call()中pthread_cond_signal()失败，返回的错误码为%d!",err);
    }

    // 检查线程是否够用，都在忙碌就写日志提醒扩容
    if(m_iThreadNum == m_iRunningThreadNum){
        time_t currtime = time(NULL);
        if(currtime - m_iLastEmgTime > 10){// 间隔10秒才写日志一次
            m_iLastEmgTime = currtime;// 记录发生时间
            Log_Error(0,"CThreadPool::Call()中发现线程池中当前空闲线程数量为0, 要考虑扩容线程池了!");
        }
    }
}